package com.own.business.log.operation.persistence.business.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.CalendarInterval;
import co.elastic.clients.elasticsearch._types.aggregations.DateHistogramBucket;
import co.elastic.clients.elasticsearch._types.aggregations.MultiBucketBase;
import co.elastic.clients.elasticsearch._types.query_dsl.ExistsQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.TermQuery;
import com.own.business.log.operation.persistence.business.config.LogOperationElasticConfig;
import com.own.business.log.operation.persistence.business.constant.LogOperationFieldConstant;
import com.own.business.log.operation.persistence.business.service.LogOperationElasticService;
import com.own.business.log.operation.persistence.entity.po.LogOperation;
import com.own.business.log.operation.persistence.entity.po.LogOperationElasticPo;
import com.own.business.log.operation.persistence.entity.query.LogOperationQuery;
import com.own.business.log.operation.persistence.entity.query.LogOperationStatsTimeRangeQuery;
import com.own.business.log.operation.persistence.entity.vo.LogOperationElasticVo;
import com.own.component.common.exception.BusinessSimpleException;
import com.own.component.common.model.ItemCountModel;
import com.own.component.common.model.ItemLongCountModel;
import com.own.component.common.model.entity.ChildrenItemModel;
import com.own.component.elastic.constant.CalendarIntervalConstant;
import com.own.component.elastic.constant.E;
import com.own.component.elastic.entity.BaseElasticPo;
import com.own.component.elastic.service.config.ElasticSearchConfig;
import com.own.component.elastic.service.impl.AbstractBaseElasticService;
import com.own.component.elastic.util.TimeRangeResolveUtil;
import com.own.component.util.time.LocalDateTimeUtil;
import com.own.constant.ConstantDateFormatPattern;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * LogOperationElasticServiceImpl
 *
 * @author chenxueli
 * @date 2023-12-15 11:42:00
 */
@Slf4j
@Service
public class LogOperationElasticServiceImpl extends AbstractBaseElasticService<
        LogOperation,
        LogOperationElasticPo,
        LogOperationElasticVo,
        LogOperationQuery
        > implements LogOperationElasticService {

    private static final String AGG_1 = "agg_1";
    private static final String AGG_2 = "agg_2";
    private static final List<ItemCountModel> EMPTY_ITEM_COUNT_MODEL_LIST_DATA = new ArrayList<>();
    @Resource
    private ElasticsearchClient elasticsearchClient;

    @PostConstruct
    private void initIndex() {
        var flag = checkIndex();
        if (flag) {
            return;
        }
        createIndex();
    }

    /**
     * 获取默认的构造信息
     *
     * @return 默认的构造信息
     */
    @Override
    public ElasticSearchConfig<LogOperation, LogOperationElasticPo, LogOperationElasticVo, LogOperationQuery> getDefaultConfig() {
        return LogOperationElasticConfig.DEFAULT_CONFIG;
    }

    /**
     * 文档名称
     *
     * @return 文档名称
     */
    @Override
    public String index() {
        return LogOperationElasticConfig.INDEX_NAME;
    }

    /**
     * 根据链路id获取日志信息
     *
     * @param traceId 链路id
     * @return 日志信息
     */
    @Override
    public LogOperationElasticVo getByTraceId(String traceId) {
        var config = ElasticSearchConfig.<
                        LogOperation,
                        LogOperationElasticPo,
                        LogOperationElasticVo,
                        LogOperationQuery
                        >builder()
                .queryBuilder((query, consumer) ->
                        b -> b.term(TermQuery.of(m -> m.field(E.buildKeyword(LogOperationElasticPo.Fields.traceId)).value(traceId)))
                )
                .sortBuilder(query -> sort -> sort.field(field ->
                        field.field(BaseElasticPo.Fields.gmtCreate).order(SortOrder.Desc)
                ))
                .build();
        return getFirst(new LogOperationQuery(), config);
    }

    /**
     * 统计时间范围内的操作次数
     *
     * @param query 查询条件
     * @return 操作次数统计
     */
    @Override
    public List<ChildrenItemModel<ItemCountModel, ItemCountModel>> statsForTimeRange(LogOperationStatsTimeRangeQuery query) {
        resolveQueryTime(query);
        // 执行信息查询
        var interval = CalendarIntervalConstant.MAP.getOrDefault(query.getInterval(), CalendarInterval.Day);
        var format = CalendarIntervalConstant.FORMAT_MAP.getOrDefault(query.getInterval(), "yyyy-MM-dd");
        var response = aggregation(query, agg -> agg
                .aggregations(AGG_1, a -> a.dateHistogram(d -> d
                        .field(LogOperationElasticPo.Fields.requestTime)
                        .calendarInterval(interval)
                        .format(format)
                        .timeZone("Asia/Shanghai")
                ))
                .terms(t -> t.field(E.buildKeyword(LogOperationElasticPo.Fields.client)))
        );
        var keyList = TimeRangeResolveUtil.resolve(interval, query.getRequestTimeStartObject(), query.getRequestTimeEndObject());
        return response.sterms()
                .buckets()
                .array()
                .stream()
                .filter(entry -> StringUtils.isNotBlank(entry.key().stringValue()))
                .map(entry -> {
                    var item = new ItemCountModel(entry.key().stringValue(), entry.docCount());
                    var time = entry.aggregations().get(AGG_1);
                    var dateCountMap = time.dateHistogram().buckets().array().stream().collect(Collectors.toMap(DateHistogramBucket::keyAsString, MultiBucketBase::docCount));
                    var list = keyList.stream()
                            .map(temp -> new ItemCountModel(temp, dateCountMap.getOrDefault(temp, 0L)))
                            .toList();
                    return new ChildrenItemModel<>(item, list);
                })
                .toList();
    }

    /**
     * 统计时间范围内的用户id操作次数
     *
     * @param query 查询条件
     * @return 操作次数统计
     */
    @Override
    public List<ChildrenItemModel<ItemCountModel, ItemLongCountModel>> statsForUserTimeRange(LogOperationStatsTimeRangeQuery query) {
        resolveQueryTime(query);
        // 执行信息查询
        var interval = CalendarIntervalConstant.MAP.getOrDefault(query.getInterval(), CalendarInterval.Day);
        var format = CalendarIntervalConstant.FORMAT_MAP.getOrDefault(query.getInterval(), "yyyy-MM-dd");
        // 执行信息查询（添加不查询的用户id）
        if (query.getExcludeUserIds() == null) {
            query.setExcludeUserIds(new HashSet<>());
        }
        query.getExcludeUserIds().add(-1L);
        var response = aggregation(query, agg -> agg
                .aggregations(AGG_1, a -> a
                        .terms(t -> t.field(LogOperationElasticPo.Fields.fkUserId).size(query.getRows()))
                )
                .dateHistogram(d -> d
                        .field(LogOperationElasticPo.Fields.requestTime)
                        .calendarInterval(interval)
                        .format(format)
                        .timeZone("Asia/Shanghai")
                )
        );
        return response.dateHistogram()
                .buckets()
                .array()
                .stream()
                .map(entry -> {
                    var item = new ItemCountModel(entry.keyAsString(), entry.docCount());
                    // 获取子节点信息
                    var userIdAgg = entry.aggregations().get(AGG_1);
                    var children = userIdAgg.lterms()
                            .buckets()
                            .array()
                            .stream()
                            .map(userEntry -> new ItemLongCountModel(userEntry.key(), (int) userEntry.docCount()))
                            .toList();
                    return new ChildrenItemModel<>(item, children);
                })
                .toList();
    }

    /**
     * 统计时间范围内的用户id数量
     *
     * @param query 查询条件
     * @return 用户id数量
     */
    @Override
    public List<ItemCountModel> statsForUserCountTimeRange(LogOperationStatsTimeRangeQuery query) {
        resolveQueryTime(query);
        // 执行信息查询
        var interval = CalendarIntervalConstant.MAP.getOrDefault(query.getInterval(), CalendarInterval.Day);
        var format = CalendarIntervalConstant.FORMAT_MAP.getOrDefault(query.getInterval(), "yyyy-MM-dd");
        // 执行信息查询（添加不查询的用户id）
        if (query.getExcludeUserIds() == null) {
            query.setExcludeUserIds(new HashSet<>());
        }
        query.getExcludeUserIds().add(-1L);
        var response = aggregation(query,
                builder -> {
                    if (query.getOnlyApp() != null && query.getOnlyApp() == 1) {
                        // 排除设备id或者系统名称不存在的数据不存在
                        builder.must(ExistsQuery.of(m -> m.field(LogOperationFieldConstant.HEADER_DEVICE_ID))._toQuery());
                        builder.must(ExistsQuery.of(m -> m.field(LogOperationFieldConstant.HEADER_SYSTEM_NAME))._toQuery());
                    }
                },
                agg -> agg.aggregations(AGG_1, a -> a
                                .cardinality(t -> t.field(LogOperationElasticPo.Fields.fkUserId))
                        )
                        .dateHistogram(d -> d
                                .field(LogOperationElasticPo.Fields.requestTime)
                                .calendarInterval(interval)
                                .format(format)
                                .timeZone("Asia/Shanghai")
                        )
        );
        var resultMap = response.dateHistogram()
                .buckets()
                .array()
                .stream()
                .map(entry -> new ItemCountModel(entry.keyAsString(), entry.aggregations().get(AGG_1).cardinality().value()))
                .collect(Collectors.toMap(ItemCountModel::getName, Function.identity()));
        var keyList = TimeRangeResolveUtil.resolve(interval, query.getRequestTimeStartObject(), query.getRequestTimeEndObject());
        return keyList
                .stream()
                .map(key -> resultMap.containsKey(key) ? resultMap.get(key) : new ItemCountModel(key, 0L))
                .toList();
    }

    /**
     * 统计时间范围内的设备数量
     *
     * @param query 查询条件
     * @return 设备数量
     */
    @Override
    public List<ChildrenItemModel<ItemCountModel, ItemCountModel>> statsForDeviceCountTimeRange(LogOperationStatsTimeRangeQuery query) {
        resolveQueryTime(query);
        // 执行信息查询
        var interval = CalendarIntervalConstant.MAP.getOrDefault(query.getInterval(), CalendarInterval.Day);
        var format = CalendarIntervalConstant.FORMAT_MAP.getOrDefault(query.getInterval(), "yyyy-MM-dd");
        var response = aggregation(query,
                builder -> {
                    // 排除设备id或者系统名称不存在的数据不存在
                    builder.must(ExistsQuery.of(m -> m.field(LogOperationFieldConstant.HEADER_DEVICE_ID))._toQuery());
                    builder.must(ExistsQuery.of(m -> m.field(LogOperationFieldConstant.HEADER_SYSTEM_NAME))._toQuery());
                },
                agg -> agg.aggregations(AGG_1, a -> a
                                .terms(t -> t
                                        .field(LogOperationFieldConstant.HEADER_SYSTEM_NAME)
                                        .size(query.getRows())
                                )
                                .aggregations(AGG_2, b -> b
                                        .cardinality(t -> t.field(LogOperationFieldConstant.HEADER_DEVICE_ID))
                                )
                        )
                        .dateHistogram(d -> d
                                .field(LogOperationElasticPo.Fields.requestTime)
                                .calendarInterval(interval)
                                .format(format)
                                .timeZone("Asia/Shanghai")
                        )
        );
        var resultMap = response.dateHistogram()
                .buckets()
                .array()
                .stream()
                .map(entry -> {
                    var item = new ItemCountModel(entry.keyAsString(), entry.docCount());
                    // 获取子节点信息
                    var systemNameAgg = entry.aggregations().get(AGG_1);
                    var children = systemNameAgg.sterms()
                            .buckets()
                            .array()
                            .stream()
                            .map(tempEntry -> new ItemCountModel(tempEntry.key().stringValue(), tempEntry.aggregations().get(AGG_2).cardinality().value()))
                            .toList();
                    return new ChildrenItemModel<>(item, children);
                })
                .collect(Collectors.toMap(item -> item.getItem().getName(), Function.identity()));
        var keyList = TimeRangeResolveUtil.resolve(interval, query.getRequestTimeStartObject(), query.getRequestTimeEndObject());
        return keyList.stream().map(key ->
                        resultMap.containsKey(key)
                                ? resultMap.get(key)
                                : new ChildrenItemModel<>(new ItemCountModel(key, 0L), EMPTY_ITEM_COUNT_MODEL_LIST_DATA)
                )
                .toList();
    }

    /**
     * 处理查询时间信息
     *
     * @param query 查询条件
     */
    private void resolveQueryTime(LogOperationQuery query) {
        // 手动处理时间信息
        var startTime = StringUtils.isBlank(query.getRequestTimeStart())
                ? null
                : LocalDateTimeUtil
                .parse(
                        query.getRequestTimeStart(),
                        ConstantDateFormatPattern.Pattern.DATETIME,
                        ConstantDateFormatPattern.Pattern.DATETIME_MM,
                        ConstantDateFormatPattern.Pattern.DATE
                );
        var endTime = StringUtils.isBlank(query.getRequestTimeEnd())
                ? null
                : LocalDateTimeUtil
                .parse(
                        query.getRequestTimeEnd(),
                        ConstantDateFormatPattern.Pattern.DATETIME,
                        ConstantDateFormatPattern.Pattern.DATETIME_MM,
                        ConstantDateFormatPattern.Pattern.DATE
                );
        // 设置时间信息
        var now = LocalDateTime.now();
        startTime = startTime == null ? now : startTime;
        endTime = endTime == null ? now : endTime;
        if (!endTime.isAfter(startTime)) {
            throw new BusinessSimpleException("时间范围不合法");
        }
        query.setRequestTimeStart(LocalDateTimeUtil.format(startTime, ConstantDateFormatPattern.Pattern.DATETIME));
        query.setRequestTimeEnd(LocalDateTimeUtil.format(endTime, ConstantDateFormatPattern.Pattern.DATETIME));
        query.setRequestTimeStartObject(startTime);
        query.setRequestTimeEndObject(endTime);
    }

}
